/**
 *  golive
 *
 *  Create by songli on 2021/05/24
 *  Copyright © 2021 居理买房网. All rights reserved.
 */

package repository

import (
	"context"
	"errors"
	"git.julive.com/julive/user/server/constant"
	"strconv"
	"sync"

	"github.com/go-redis/redis/v8"
	"go.uber.org/zap"
	"gorm.io/gorm"

	"git.julive.com/julive/user/server/model"
	"git.julive.com/micro/pkg/dao"
	"git.julive.com/micro/pkg/logx"
	"git.julive.com/micro/pkg/redisx"
	"git.julive.com/micro/pkg/util"
)

type UserRepository interface {
	dao.Dao

	CreateUser(ctx context.Context, m *model.User) error

	GetUserById(ctx context.Context, id int32) (*model.User, error)
	FindUserById(ctx context.Context, id int32) (*model.User, error)
	GetUserList(ctx context.Context, status, lastid, pagesize, page int32) ([]*model.User, int, error)

	UpdateUserStatus(ctx context.Context, id, status int32) (int64, error)
	UpdateUserCount(ctx context.Context, id, num int32, column string) (int64, error)

	DeleteUserById(ctx context.Context, id int32) (int64, error)
}

type userRepository struct {
	dao.Dao

	logger *zap.Logger
}

// NewUserRepository 创建User存储库实例
func NewUserRepository() UserRepository {
	logger := logx.Logger.With(zap.String("module", "user/repository"), zap.String("svc", "user"))
	repo := &userRepository{
		dao.NewRepository(),
		logger,
	}
	return repo
}

func (repo *userRepository) CreateUser(ctx context.Context, m *model.User) error {
	if err := repo.WriteDB("hr").Create(m).Error; err != nil {
		return err
	}
	key := util.AppendString("jl_user_", strconv.Itoa(int(m.Id)))
	expire := option.CacheMinute5 + util.RandDuration(120)
	redisx.SetHashTable(ctx, repo.Redis(), key, m, expire)

	return nil
}

func (repo *userRepository) GetUserById(ctx context.Context, id int32) (*model.User, error) {
	m := &model.User{}
	key := util.AppendString("jl_user_", strconv.Itoa(int(id)))
	err := redisx.HGetAll(ctx, repo.Redis(), key, m)
	if err == nil {
		if m.Id == 0 {
			return nil, nil
		}
		return m, nil
	}

	m, err = repo.FindUserById(ctx, id)
	if err != nil {
		return nil, err
	}

	if m != nil {
		expire := option.CacheMinute5 + util.RandDuration(120)
		redisx.SetHashTable(ctx, repo.Redis(), key, m, expire)
		return m, nil
	} else {
		m = &model.User{Id: 0}
		expire := option.CacheMinute1 + util.RandDuration(30)
		redisx.SetHashTable(ctx, repo.Redis(), key, m, expire)
		return nil, nil
	}
}

func (repo *userRepository) FindUserById(ctx context.Context, id int32) (*model.User, error) {
	m := &model.User{}
	err := repo.ReadDB("hr").Where("id = ?", id).First(m).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, nil
		}
		return nil, err
	}
	return m, nil
}

func (repo *userRepository) GetUsersCount(ctx context.Context, status int32) (int64, error) {
	logger := repo.logger.With(zap.String("func", "GetUsersCount"))

	key := util.AppendString("jl_usercnt_", strconv.Itoa(int(status)))
	cnt, err := redisx.GetNumber(ctx, repo.Redis(), key)
	if err == nil {
		return cnt, nil
	}
	cnt, err = repo.LoadUsersCount(ctx, status)
	if err != nil {
		return 0, err
	}
	err = repo.Redis().Set(ctx, key, cnt, constant.CacheMinute5).Err()
	if err != nil {
		logger.Error("redis.Set", zap.String("key", key), zap.Error(err))
	}
	return cnt, nil
}

func (repo *userRepository) LoadUsersCount(ctx context.Context, status int32) (int64, error) {
	var count int64
	tx := repo.ReadDB("hr").Model(&model.User{}).Select("count(id)")
	tx = tx.Where("status=?", status)
	if err := tx.Count(&count).Error; err != nil {
		return 0, err
	}
	return count, nil
}

func (repo *userRepository) GetUserList(ctx context.Context, status, lastid, pagesize, page int32) ([]*model.User, int, error) {
	logger := repo.logger.With(zap.String("func", "GetUserList"))
	ids, cnt, err := repo.GetUserListIds(ctx, status, lastid, pagesize, page)
	if err != nil {
		return nil, 0, err
	}

	ctx1, cancel := context.WithTimeout(ctx, constant.CRequestTimeout)
	defer cancel()

	users, err := repo.GetUserList4Concurrent(ctx1, ids, repo.GetUserById)
	logger.Debug("GetUserList4Concurrent", zap.Any("users", users), zap.Error(err))
	if err != nil {
		return nil, 0, err
	}
	return users, cnt, nil
}

func (repo *userRepository) GetUserListIds(ctx context.Context, status, lastid, pagesize, page int32) ([]int32, int, error) {
	key := util.AppendString("jl_user_ids_", strconv.Itoa(int(status)))
	ids, cnt, err := redisx.ZRevRangeWithCard(ctx, repo.Redis(), key, lastid, pagesize, page)
	if err == nil {
		return ids, cnt, nil
	}

	ids, args, err := repo.FindUserListIds(ctx, status, lastid, pagesize, page)
	if err != nil {
		return nil, 0, err
	}
	expire := constant.CacheMinute5 + util.RandDuration(120)
	redisx.SetSortedSet(ctx, repo.Redis(), key, args, expire)
	return ids, len(args), nil
}

func (repo *userRepository) FindUserListIds(ctx context.Context, status, lastid, pagesize, page int32) ([]int32, []*redis.Z, error) {
	tx := repo.ReadDB("hr").Model(&model.User{}).Select("id")
	tx = tx.Where("status=?", status)
	tx = tx.Order("id DESC")
	rows, err := tx.Rows()
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return []int32{}, []*redis.Z{}, nil
		}
		return nil, nil, err
	}
	defer rows.Close()

	var ids []int32
	args := []*redis.Z{}
	for rows.Next() {
		var (
			id int32
		)
		err = rows.Scan(&id)
		if err != nil {
			return nil, nil, err
		}

		check := false
		if lastid == 0 {
			check = true
		} else if lastid > id {
			check = true
		}
		if check {
			if len(ids) < int(pagesize) {
				ids = append(ids, id)
			}
		}
		args = append(args, &redis.Z{float64(id), id})
	}
	if err = rows.Err(); err != nil {
		return nil, nil, err
	}
	return ids, args, nil
}

func (repo *userRepository) GetUserList4Concurrent(ctx context.Context, ids []int32, fn func(context.Context, int32) (*model.User, error)) ([]*model.User, error) {
	var wg sync.WaitGroup

	count := len(ids)
	outputs := make([]*concurrentUserOutput, count)
	wg.Add(count)

	for idx, id := range ids {
		go func(idx int, id int32, wg *sync.WaitGroup) {
			defer wg.Done()
			user, err := fn(ctx, id)
			outputs[idx] = &concurrentUserOutput{
				object: user,
				err:    err,
			}
		}(idx, id, &wg)
	}
	wg.Wait()

	users := make([]*model.User, 0, count)
	for _, output := range outputs {
		if output.err == nil && output.object != nil {
			users = append(users, output.object)
		}
	}
	return users, nil
}

type concurrentUserOutput struct {
	object *model.User
	err    error
}

func (repo *userRepository) UpdateUserStatus(ctx context.Context, id, status int32) (int64, error) {
	logger := repo.logger.With(zap.String("func", "UpdateUserStatus"))
	logger.Debug("invoke info", zap.Int32("id", id), zap.Int32("status", status))
	tx := repo.WriteDB("hr").Model(model.User{}).Where("id = ?", id)
	tx = tx.Update("status", status)
	if tx.Error != nil {
		logger.Error("tx.Update error", zap.Error(tx.Error))
		return 0, tx.Error
	}
	key := util.AppendString("jl_user_", strconv.Itoa(int(id)))
	reply, err := repo.Redis().Del(ctx, key).Result()
	if err != nil {
		logger.Warn("Del Cache", zap.String("key", key), zap.Int64("reply", reply), zap.Error(err))
	}
	return tx.RowsAffected, nil
}

func (repo *userRepository) UpdateUserCount(ctx context.Context, id, num int32, column string) (int64, error) {
	logger := repo.logger.With(zap.String("func", "UpdateUserCount"))
	logger.Debug("invoke info", zap.Int32("id", id), zap.Int32("num", num), zap.String("column", column))
	tx := repo.WriteDB("hr").Model(model.User{}).Where("id = ?", id)
	tx = tx.Updates(map[string]interface{}{column: gorm.Expr(column+" + ?", num)})
	if tx.Error != nil {
		logger.Error("tx.Updates error", zap.Error(tx.Error))
		return 0, tx.Error
	}
	key := util.AppendString("jl_user_", strconv.Itoa(int(id)))
	reply, err := repo.Redis().Del(ctx, key).Result()
	if err != nil {
		logger.Warn("Del Cache", zap.String("key", key), zap.Int64("reply", reply), zap.Error(err))
	}
	return tx.RowsAffected, nil
}

func (repo *userRepository) DeleteUserById(ctx context.Context, id int32) (int64, error) {
	logger := repo.logger.With(zap.String("func", "DeleteUserById"))
	logger.Debug("invoke info", zap.Int32("id", id))
	tx := repo.WriteDB("hr").Delete(&model.User{}, id)
	if tx.Error != nil {
		logger.Error("tx.Delete error", zap.Error(tx.Error))
		return 0, tx.Error
	}
	key := util.AppendString("jl_user_", strconv.Itoa(int(id)))
	reply, err := repo.Redis().Del(ctx, key).Result()
	logger.Debug("Del Cache", zap.String("key", key), zap.Int64("reply", reply), zap.Error(err))

	status := []int{0, 1}
	for _, s := range status {
		key := util.AppendString("jl_user_ids_", strconv.Itoa(s))
		err := repo.Redis().ZRem(ctx, key, id).Err()
		if err != nil {
			logger.Warn("pipeliner.ZRem", zap.String("key", key), zap.Int32("id", id), zap.Error(err))
		}
	}

	return tx.RowsAffected, nil
}
